package org.yonggan.spark;

import akka.dispatch.Foreach;
import com.alibaba.fastjson.JSONObject;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import org.yonggan.conf.ConfigurationManager;
import org.yonggan.constant.Constants;
import org.yonggan.dao.ISessionAggrStatDAO;
import org.yonggan.dao.ISessionDetailDAO;
import org.yonggan.dao.ISessionRandomExtractDAO;
import org.yonggan.dao.ITaskDAO;
import org.yonggan.dao.factory.DAOFactory;
import org.yonggan.domain.SessionAggrStat;
import org.yonggan.domain.SessionDetail;
import org.yonggan.domain.SessionRandomExtract;
import org.yonggan.domain.Task;
import org.yonggan.mock.MockData;
import org.yonggan.util.*;
import scala.Tuple2;

import java.util.*;

/**
 * 用户访问session分析Spark作业
 * <p>
 * 接收用户创建的分析任务，用户可能指定的条件如下：
 * <p>
 * 1、时间范围：起始日期~结束日期
 * 2、性别：男或女
 * 3、年龄范围
 * 4、职业：多选
 * 5、城市：多选
 * 6、搜索词：多个搜索词，只要某个session中的任何一个action搜索过指定的关键词，那么session就符合条件
 * 7、点击品类：多个品类，只要某个session中的任何一个action点击过某个品类，那么session就符合条件
 * <p>
 * 我们的spark作业如何接受用户创建的任务？
 * <p>
 * J2EE平台在接收用户创建任务的请求之后，会将任务信息插入MySQL的task表中，任务参数以JSON格式封装在task_param
 * 字段中
 * <p>
 * 接着J2EE平台会执行我们的spark-submit shell脚本，并将taskid作为参数传递给spark-submit shell脚本
 * spark-submit shell脚本，在执行时，是可以接收参数的，并且会将接收的参数，传递给Spark作业的main函数
 * 参数就封装在main函数的args数组中
 */
public class UserVisitSessionAnalyzeSpark {

    public static void main(String[] args) {

        args = new String[]{"1"};

        SparkConf conf = new SparkConf();
        conf.setAppName(Constants.SPARK_APP_NAME_SESSION);
        conf.setMaster("local[*]");

        JavaSparkContext sc = new JavaSparkContext(conf);

        SQLContext sqlContext = getSQLContext(sc.sc());
        MockData.mock(sc, sqlContext);

        // 创建需要使用的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        // 创建需要的Dao
        Long taskId = ParamUtils.getTaskIdFromArgs(args,
                Constants.SPARK_LOCAL_TASKID_SESSION);

        Task task = taskDAO.findById(taskId);

        if (task == null) {
            System.out.println(new Date() + ": cannot find this task with id [" + taskId + "].");
            return;
        }
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());


        // 提取用户行为数据
        JavaRDD<Row> actionRDD = SparkUtils.getActionRDDByDateRange(sqlContext, taskParam);
//        actionRDD.persist()

        /**
         * 提取到用户行为数据 做成 (sessionid,row)
         */
        final JavaPairRDD<String, Row> session2ActionRDD = getSession2ActionRDD(actionRDD);

        // 首先将数据行为数据聚合 按照session_id进行groupByKey分组
        // 此时的数据的粒度就是session粒度了，然后呢，可以将session粒度的数据
        // 与用户信息数据，进行join
        // 然后就可以获取到session粒度的数据，同时呢，数据里面还包含了session对应的user的信息
        // 到这里为止，获取的数据是<sessionid,(sessionid,searchKeywords,clickCategoryIds,age,professional,city,sex)>
        JavaPairRDD<String, String> fullAggrInfo = aggregateBySession(sqlContext, actionRDD);

        /**
         * 正对与session 粒度的聚合数据  需要按照指定的筛选出的指定的指标
         */
        System.out.println(fullAggrInfo.count());
        for (Tuple2<String, String> t : fullAggrInfo.take(10)) {
            System.out.println(t._2);
        }


        Accumulator<String> sessionAggrSta = sc.accumulator("", new SessionAggrStatusAccumulator());


        JavaPairRDD<String, String> filterRdd = filterSessionAndAggrStatus(fullAggrInfo, taskParam, sessionAggrSta);
        System.out.println(filterRdd.count());


//        执行随机抽取session
        randomExtractSession(filterRdd,taskId,session2ActionRDD);

        /**
         * session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
         *
         * 如果不进行重构，直接来实现，思路：
         * 1、actionRDD，映射成<sessionid,Row>的格式
         * 2、按sessionid聚合，计算出每个session的访问时长和访问步长，生成一个新的RDD
         * 3、遍历新生成的RDD，将每个session的访问时长和访问步长，去更新自定义Accumulator中的对应的值
         * 4、使用自定义Accumulator中的统计值，去计算各个区间的比例
         * 5、将最后计算出来的结果，写入MySQL对应的表中
         *
         * 普通实现思路的问题：
         * 1、为什么还要用actionRDD，去映射？其实我们之前在session聚合的时候，映射已经做过了。多此一举
         * 2、是不是一定要，为了session的聚合这个功能，单独去遍历一遍session？其实没有必要，已经有session数据
         * 		之前过滤session的时候，其实，就相当于，是在遍历session，那么这里就没有必要再过滤一遍了
         *
         * 重构实现思路：
         * 1、不要去生成任何新的RDD（处理上亿的数据）
         * 2、不要去单独遍历一遍session的数据（处理上千万的数据）
         * 3、可以在进行session聚合的时候，就直接计算出来每个session的访问时长和访问步长
         * 4、在进行过滤的时候，本来就要遍历所有的聚合session信息，此时，就可以在某个session通过筛选条件后
         * 		将其访问时长和访问步长，累加到自定义的Accumulator上面去
         * 5、就是两种截然不同的思考方式，和实现方式，在面对上亿，上千万数据的时候，甚至可以节省时间长达
         * 		半个小时，或者数个小时
         *
         * 开发Spark大型复杂项目的一些经验准则：
         *
         * 1、尽量少生成RDD
         * 2、尽量少对RDD进行算子操作，如果有可能，尽量在一个算子里面，实现多个需要做的功能
         * 3、尽量少对RDD进行shuffle算子操作，比如groupByKey、reduceByKey、sortByKey（map、mapToPair）
         * 		shuffle操作，会导致大量的磁盘读写，严重降低性能
         * 		有shuffle的算子，和没有shuffle的算子，甚至性能，会达到几十分钟，甚至数个小时的差别
         * 		有shfufle的算子，很容易导致数据倾斜，一旦数据倾斜，简直就是性能杀手（完整的解决方案）
         *
         * 4、无论做什么功能，性能第一
         * 		在传统的J2EE或者.NET后者PHP，软件/系统/网站开发中，我认为是架构和可维护性，可扩展性的重要
         * 		程度，远远高于了性能，大量的分布式的架构，设计模式，代码的划分，类的划分（高并发网站除外）
         *
         * 		在大数据项目中，比如MapReduce、Hive、Spark、Storm，我认为性能的重要程度，远远大于一些代码
         * 		的规范，和设计模式，代码的划分，类的划分；大数据，大数据，最重要的，就是性能
         * 		主要就是因为大数据以及大数据项目的特点，决定了，大数据的程序和项目的速度，都比较慢
         * 		如果不优先考虑性能的话，会导致一个大数据处理程序运行时间长度数个小时，甚至数十个小时
         * 		此时，对于用户体验，简直就是一场灾难
         *
         * 		所以，推荐大数据项目，在开发和代码的架构中，优先考虑性能；其次考虑功能代码的划分、解耦合
         *
         * 		我们如果采用第一种实现方案，那么其实就是代码划分（解耦合、可维护）优先，设计优先
         * 		如果采用第二种方案，那么其实就是性能优先
         *
         */

        //accumulator统计结果写入mysql
        calculateAndPersistAggrStat(sessionAggrSta.value(), taskId);

        sc.stop();

    }

    /**
     * 将用户有的基础映射成为<String,row>
     * @param actionRDD
     * @return
     */
    private static JavaPairRDD<String,Row> getSession2ActionRDD(JavaRDD<Row> actionRDD) {

      return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
          @Override
          public Tuple2<String, Row> call(Row row) throws Exception {
              final String sessionId = row.getString(2);
              return new Tuple2<String, Row>(sessionId,row);
          }
      });


    }

    /**
     * 随机session 取样
     *
     * @param filterRdd
     */
    private static void randomExtractSession(JavaPairRDD<String, String> filterRdd,
                                     final long taskid,JavaPairRDD<String, Row> sessionid2actionRDD) {

        // 第一步，计算出每小时的session数量（yyyy-mm dd ）
        JavaPairRDD<String, String> time2Session = filterRdd.mapToPair(new PairFunction<Tuple2<String, String>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<String, String> tuple) throws Exception {

                String argsInfo = tuple._2;
                // 提取出时间
                String time = StringUtils.getFieldFromConcatString(argsInfo, "[|]", Constants.FIELD_START_TIME);
                // 时间 yyyy-MM-dd
                String hour = DateUtils.getDateHour(time);

                return new Tuple2<String, String>(hour, argsInfo);
            }
        });

        /**
         * 实现随机抽取
         */
        // 得到每小时的session数量
        Map<String, Object> countByMap = time2Session.countByKey();
        // 按照时间比例 随机抽取每天每小时抽取的session索引

        final Random random = new Random();

        Map<String, Map<String, Long>> dayHourCountMap = new HashMap
                <String, Map<String, Long>>();
        for (Map.Entry<String, Object> entry : countByMap.entrySet()) {

            // 时间
            String dayHour = entry.getKey();
            String[] split = dayHour.split("-");
            String day = split[0];
            String hour = split[1];

            // 总数
            long cnt = Long.valueOf(String.valueOf(entry.getValue()));

            Map<String, Long> hourCountMap = dayHourCountMap.get(dayHour);

            if (hourCountMap == null) {
                hourCountMap = new HashMap<String, Long>();
                dayHourCountMap.put(dayHour, hourCountMap);
            }
            // 存放每小时
            hourCountMap.put(hour, cnt);
        }
        // 开始按照时间随机抽取session
        // 总需要100个session 先按照天数进行平分
        long extraNumberPerDay = 100 / dayHourCountMap.size();

        // 每天每小时sesion 的索引
        final Map<String, Map<String, List<Integer>>> dayHourExtraMap = new HashMap<String, Map<String, List<Integer>>>();

        for (Map.Entry<String, Map<String, Long>> entry :
                dayHourCountMap.entrySet()) {

            final String date = entry.getKey();
            final Map<String, Long> hourCountMap = entry.getValue();

            // 计算这天的sessoion 总数
            long sessionCount = 0L;
            for (Long hourCount : hourCountMap.values()) {
                sessionCount += hourCount;
            }

            // 遍历每小时
            for (Map.Entry<String, Long> hourCountEntry : hourCountMap.entrySet()) {

                String hour = hourCountEntry.getKey();
                Long cout = hourCountEntry.getValue();

                // 先获取这天对应的随机数
                final Map<String, List<Integer>> listMap = dayHourExtraMap.get(date);
                if (listMap == null) {
                    final Map<String, List<Integer>> map = new HashMap<String, List<Integer>>();
                    dayHourExtraMap.put(date, map);
                }

                // 每小时的session 占据每天的 数量的比例
                long hourExtraNumber = (long) ((double) cout / (double) sessionCount * extraNumberPerDay);

                if (hourExtraNumber > cout) { // 如果說 需要抽取的數量大于 总数
                    hourExtraNumber = cout;
                }

                // 生成上随机范围的随机数
                final List<Integer> extraIndexLst = listMap.get(hour);
                if (extraIndexLst == null) {
                    listMap.put(hour, new ArrayList<Integer>());
                }
                // 生成索引
                for (int i = 0; i < hourExtraNumber; i++) {

                    int idx = random.nextInt(Integer.valueOf(String.valueOf(cout)));
                    while (extraIndexLst.contains(idx)) { // 保证不能是重复的
                        idx = random.nextInt(Integer.valueOf(String.valueOf(cout)));
                    }
                    // 添加生成的随机索引
                    extraIndexLst.add(idx);

                }
            }
        }

        /**
         * 第三步 根据随机索引 抽取每天每小时的数据
         *
         * 遍历
         *  [dayhour,(sesionid-argsInfo .... )]
         *
         */
        final JavaPairRDD<String, Iterable<String>> groupedRDD = time2Session.groupByKey();

        JavaPairRDD<String, String> extractSessionidsRDD =   groupedRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<String, Iterable<String>>, String, String>() {

            @Override
            public Iterable<Tuple2<String, String>> call(Tuple2<String, Iterable<String>> tuple) throws Exception {

                // 存放返回 SessionId
                List<Tuple2<String, String>> extraSessionIds = new ArrayList<Tuple2<String, String>>();

                final String dayHour = tuple._1;
                final String day = dayHour.split("-")[0];
                final String hour = dayHour.split("-")[1];


                // 随机抽取的索引
                final List<Integer> extraRandomIdx = dayHourExtraMap.get(day).get(hour);


                int index = 0;
                final Iterator<String> itor = tuple._2.iterator();

                final ISessionRandomExtractDAO sessionRandomExtractDAO = DAOFactory.getSessionRandomExtractDAO();

                while (itor.hasNext()) {

                    final String sesionAggrInfo = itor.next();
                    if (extraRandomIdx.contains(sesionAggrInfo)) { // 写MySQL 返回 sessionId

                        String sessionid = StringUtils.getFieldFromConcatString(
                                sesionAggrInfo, "\\|", Constants.FIELD_SESSION_ID);

                        // 将数据写入MySQL
                        SessionRandomExtract sessionRandomExtract = new SessionRandomExtract();
                        sessionRandomExtract.setTaskid(taskid);
                        sessionRandomExtract.setSessionid(sessionid);
                        sessionRandomExtract.setStartTime(StringUtils.getFieldFromConcatString(
                                sesionAggrInfo, "\\|", Constants.FIELD_START_TIME));
                        sessionRandomExtract.setSearchKeywords(StringUtils.getFieldFromConcatString(
                                sesionAggrInfo, "\\|", Constants.FIELD_SEARCH_KEYWORDS));
                        sessionRandomExtract.setClickCategoryIds(StringUtils.getFieldFromConcatString(
                                sesionAggrInfo, "\\|", Constants.FIELD_CLICK_CATEGORY_IDS));

                        // 写入
                        sessionRandomExtractDAO.insert(sessionRandomExtract);

                        extraSessionIds.add(new Tuple2<String, String>(sessionid, sessionid));
                    }
                    index++;
                }
                return extraSessionIds;
            }
        });

        /**
         * 第四步：获取抽取出来的session的明细数据
         */
        JavaPairRDD<String, Tuple2<String, Row>> extractSessionDetailRDD =
                extractSessionidsRDD.join(sessionid2actionRDD);

        extractSessionDetailRDD.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Tuple2<String, Row>>>>() {
            @Override
            public void call(Iterator<Tuple2<String, Tuple2<String, Row>>> iterator) throws Exception {

                List<SessionDetail> sessionDetails = new ArrayList<SessionDetail>();

                while(iterator.hasNext()) {
                    Tuple2<String, Tuple2<String, Row>> tuple = iterator.next();

                    Row row = tuple._2._2;

                    SessionDetail sessionDetail = new SessionDetail();
                    sessionDetail.setTaskid(taskid);
                    sessionDetail.setUserid(row.getLong(1));
                    sessionDetail.setSessionid(row.getString(2));
                    sessionDetail.setPageid(row.getLong(3));
                    sessionDetail.setActionTime(row.getString(4));
                    sessionDetail.setSearchKeyword(row.getString(5));
                    sessionDetail.setClickCategoryId(row.getLong(6));
                    sessionDetail.setClickProductId(row.getLong(7));
                    sessionDetail.setOrderCategoryIds(row.getString(8));
                    sessionDetail.setOrderProductIds(row.getString(9));
                    sessionDetail.setPayCategoryIds(row.getString(10));
                    sessionDetail.setPayProductIds(row.getString(11));

                    sessionDetails.add(sessionDetail);
                }

                ISessionDetailDAO sessionDetailDAO = DAOFactory.getSessionDetailDAO();
                sessionDetailDAO.insertBatch(sessionDetails);


            }
        });


    }

    public static SQLContext getSQLContext(SparkContext sc) {

        Boolean bool = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);

        if (bool) {
            return new SQLContext(sc); // 本地环境
        } else {
            return new HiveContext(sc); // 生产环境
        }
    }



    /**
     * 对行为数据按session粒度进行聚合
     *
     * @param actionRDD 行为数据RDD
     * @return session粒度聚合数据
     */
    private static JavaPairRDD<String, String> aggregateBySession(
            SQLContext sqlContext,
            JavaRDD<Row> actionRDD) {
        // Row 一行用戶的访问数据
        // 先将这个Row 转换成为<SessionId,Row>格式
        JavaPairRDD<String, Row> sessionId2ActionRDD = actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) throws Exception {
                return new Tuple2<String, Row>(row.getString(2), row);
            }
        });

        // 对行为数据分组
        JavaPairRDD<String, Iterable<Row>> sessionId2ActionRDDGrouped =
                sessionId2ActionRDD.groupByKey();

        // 每一个session 分组进行聚合session 中对搜索词与点击品类进行组合

        JavaPairRDD<Long, String> sessionAgreeParitAction = sessionId2ActionRDDGrouped.mapToPair(
                new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
                    @Override
                    public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {

                        String sessionId = tuple._1;
                        Iterable<Row> rows = tuple._2;
                        // 实现聚合
                        Iterator<Row> itor = rows.iterator();
                        StringBuffer keywordBuff = new StringBuffer();
                        StringBuffer catagoryIdBuff = new StringBuffer();

                        Long userId = null;
                        // 访问开始时间
                        Date startTime = null;
                        Date endTime = null;

                        // 访问的步长
                        int stepLength = 0;


                        // 遍历session 所有访问行为
                        while (itor.hasNext()) {
                            Row next = itor.next();
                            String searchKeyWord = next.getString(5);
                            Long categoryId = next.getLong(6);

                            if (userId == null) {
                                userId = next.getLong(1);
                            }

                            // 实际上这里要对数据说明一下
                            // 并不是每一行访问行为都有searchKeyword何clickCategoryId两个字段的
                            // 其实，只有搜索行为，是有searchKeyword字段的
                            // 只有点击品类的行为，是有clickCategoryId字段的
                            // 所以，任何一行行为数据，都不可能两个字段都有，所以数据是可能出现null值的

                            // 我们决定是否将搜索词或点击品类id拼接到字符串中去
                            // 首先要满足：不能是null值
                            // 其次，之前的字符串中还没有搜索词或者点击品类id

                            if (StringUtils.isNotEmpty(searchKeyWord) && !keywordBuff.toString().contains(searchKeyWord)) {
                                keywordBuff.append(searchKeyWord + ",");
                            }

                            if (categoryId != null && !catagoryIdBuff.toString().contains(String.valueOf(categoryId))) {
                                catagoryIdBuff.append(categoryId + ",");
                            }
//                            2018-05-31 00:00:00
                            String s = " 2018-05-31 00:00:00";
                            if (!next.getAs("action_time").toString().equals("")) {
                                s = next.getAs("action_time").toString();
                            }
                            // 计算session开始和结束时间
                            Date actionTime = DateUtils.parseTime(s);

                            if (startTime == null) {
                                startTime = actionTime;
                            }
                            if (endTime == null) {
                                endTime = actionTime;
                            }

                            if (actionTime.before(startTime)) {
                                startTime = actionTime;
                            }
                            if (actionTime.after(endTime)) {
                                endTime = actionTime;
                            }

                            // 计算session访问步长
                            stepLength++;
                        }
                        // 计算session访问时长（秒）
                        long visitLength = (endTime.getTime() - startTime.getTime()) / 1000;

                        String searchKeyWord = StringUtils.trimComma(keywordBuff.toString());
                        String catagoryId = StringUtils.trimComma(catagoryIdBuff.toString());

                        //返回 <String ,pariInfo> sessionId 中并没有 userId
                        // 聚合数据格式 key=value |  key=value
                        String partAggrInfo = Constants.FIELD_SESSION_ID + "=" + sessionId + "|"
                                + Constants.FIELD_SEARCH_KEYWORDS + "=" + searchKeyWord + "|"
                                + Constants.FIELD_CATEGORY_ID + "=" + catagoryId + "|"
                                + Constants.FIELD_VISIT_LENGTH + "=" + visitLength + "|"
                                + Constants.FIELD_STEP_LENGTH + "=" + stepLength + "|"
                                + Constants.FIELD_START_TIME + "=" + DateUtils.formatTime(startTime);// 开始时间

                        return new Tuple2<Long, String>(userId, partAggrInfo);
                    }
                });

        // 查询用户i信息

        String sql = "select * from user_info";
        JavaRDD<Row> userInfo = sqlContext.sql(sql).javaRDD();

        JavaPairRDD<Long, Row> longRowJavaPairRDD = userInfo.mapToPair(new PairFunction<Row, Long, Row>() {
            @Override
            public Tuple2<Long, Row> call(Row row) throws Exception {
                return new Tuple2<Long, Row>(row.getLong(0), row);
            }
        });

        // 将session粒度聚合数据与用户信息的数据进行jion
        // Long, Tuple2<String, Row>---> 行为信息 用户数据
        JavaPairRDD<Long, Tuple2<String, Row>> jsonPairRDD =
                sessionAgreeParitAction.join(longRowJavaPairRDD);
        //将整合到数据进行的组装成为<sessionId ,fullAggrInfo>
        JavaPairRDD<String, String> fullAggrInfo = jsonPairRDD.mapToPair(new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> tuple) throws Exception {
                // 行为数据
                String partAggrInfo = tuple._2._1;
                Row userInfoRow = tuple._2._2;

                String sessionid = StringUtils.getFieldFromConcatString(
                        partAggrInfo, "[|]", Constants.FIELD_SESSION_ID);

                // 用户的信息
//                2、性别：男或女
//                3、年龄范围
//                4、职业：多选
//                5、城市：多选
                int age = userInfoRow.getInt(3);
                String professional = userInfoRow.getString(4);
                String city = userInfoRow.getString(5);
                String grander = userInfoRow.getString(6);

                // 组合数据
                String fullAggrInfo = partAggrInfo + "|"
                        + Constants.FIELD_AGE + "=" + age + "|"
                        + Constants.FIELD_PROFESSIONAL + "=" + professional + "|"
                        + Constants.FIELD_CITY + "=" + city + "|"
                        + Constants.FIELD_SEX + "=" + grander;
                return new Tuple2<String, String>(sessionid, fullAggrInfo);
            }
        });

        return fullAggrInfo;
    }


    /**
     * 过滤session
     * 并進行
     *
     * @param session2AggrRDD
     * @return
     */
    private static JavaPairRDD<String, String> filterSessionAndAggrStatus(
            JavaPairRDD<String, String> session2AggrRDD,
            final JSONObject taskParams,
            final Accumulator<String> sessionAggrAccumulator) {
        // 提取开始年龄
        final String startAge = ParamUtils.getParam(taskParams, Constants.PARAM_START_AGE);
        final String endAge = ParamUtils.getParam(taskParams, Constants.PARAM_END_AGE);
        // 职业
        final String professional = ParamUtils.getParam(taskParams, Constants.PARAM_PROFESSIONALS);
        // 城市
        final String city = ParamUtils.getParam(taskParams, Constants.PARAM_CITIES);
        // 性别
        final String sex = ParamUtils.getParam(taskParams, Constants.PARAM_SEX);

        final String keywords = ParamUtils.getParam(taskParams, Constants.PARAM_KEYWORDS);

        final String category = ParamUtils.getParam(taskParams, Constants.PARAM_CATEGORY_IDS);

        // 拼接条件
        String _paramter = startAge != null ? Constants.PARAM_START_AGE + "=" + startAge + "|" : "";
        _paramter += endAge != null ? Constants.PARAM_END_AGE + "=" + endAge + "|" : "";
        _paramter += professional != null ? Constants.PARAM_PROFESSIONALS + "=" + professional + "|" : "";
        _paramter += city != null ? Constants.PARAM_CITIES + "=" + sex + "|" : "";
        _paramter += sex != null ? Constants.PARAM_SEX + "=" + sex + "|" : "";
        _paramter += keywords != null ? Constants.PARAM_KEYWORDS + "=" + keywords + "|" : "";
        _paramter += category != null ? Constants.PARAM_CATEGORY_IDS + "=" + category + "|" : "";

        // 处理 最后 条件式 "|" 结尾
//        if (_paramter.trim().endsWith("[|]")) {
        _paramter = _paramter.substring(0, _paramter.length() - 1);
//        }

        final String paramter = _paramter;

        // 过滤
        return session2AggrRDD.filter(new Function<Tuple2<String, String>, Boolean>() {

            @Override
            public Boolean call(Tuple2<String, String> tuple) throws Exception {
                // 提取过滤数据
                String arrgsInfo = tuple._2;
                // 按照年龄过滤 (startAge endAge)
                if (!ValidUtils.between(arrgsInfo, Constants.FIELD_AGE, paramter,
                        Constants.PARAM_START_AGE, Constants.PARAM_END_AGE)) {
                    return false;
                }
                // 按照职业过滤
                if (!ValidUtils.in(arrgsInfo, Constants.FIELD_PROFESSIONAL, paramter,
                        Constants.PARAM_PROFESSIONALS)) {
                    return false;
                }
                // 按照城市
                if (!ValidUtils.in(arrgsInfo, Constants.FIELD_CITY, paramter,
                        Constants.PARAM_CITIES)) {
                    return false;
                }
                // 性别过滤
                if (!ValidUtils.equal(arrgsInfo, Constants.FIELD_SEX, paramter,
                        Constants.PARAM_SEX)) {
                    return false;
                }
                // 搜索词
                if (!ValidUtils.in(arrgsInfo, Constants.FIELD_SEARCH_KEYWORDS, paramter,
                        Constants.PARAM_KEYWORDS)) {
                    return false;
                }
                // 种类
                if (!ValidUtils.in(arrgsInfo, Constants.FIELD_CATEGORY_ID, paramter,
                        Constants.PARAM_CATEGORY_IDS)) {
                    return false;
                }
                // 如果滿足上述条件 实现过滤
                sessionAggrAccumulator.add(Constants.SESSION_COUNT);

                // 计算出session的访问时长和访问步长的范围，并进行相应的累加
                long visitLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                        arrgsInfo, "\\|", Constants.FIELD_VISIT_LENGTH));
                long stepLength = Long.valueOf(StringUtils.getFieldFromConcatString(
                        arrgsInfo, "\\|", Constants.FIELD_STEP_LENGTH));
                calculateVisitLength(visitLength);
                calculateStepLength(stepLength);

                return true;
            }

            /**
             * 计算访问时长范围
             * @param visitLength
             */
            private void calculateVisitLength(long visitLength) {
                if (visitLength >= 1 && visitLength <= 3) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_1s_3s);
                } else if (visitLength >= 4 && visitLength <= 6) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_4s_6s);
                } else if (visitLength >= 7 && visitLength <= 9) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_7s_9s);
                } else if (visitLength >= 10 && visitLength <= 30) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_10s_30s);
                } else if (visitLength > 30 && visitLength <= 60) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_30s_60s);
                } else if (visitLength > 60 && visitLength <= 180) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_1m_3m);
                } else if (visitLength > 180 && visitLength <= 600) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_3m_10m);
                } else if (visitLength > 600 && visitLength <= 1800) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_10m_30m);
                } else if (visitLength > 1800) {
                    sessionAggrAccumulator.add(Constants.TIME_PERIOD_30m);
                }
            }

            /**
             * 计算访问步长范围
             * @param stepLength
             */
            private void calculateStepLength(long stepLength) {
                if (stepLength >= 1 && stepLength <= 3) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_1_3);
                } else if (stepLength >= 4 && stepLength <= 6) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_4_6);
                } else if (stepLength >= 7 && stepLength <= 9) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_7_9);
                } else if (stepLength >= 10 && stepLength <= 30) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_10_30);
                } else if (stepLength > 30 && stepLength <= 60) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_30_60);
                } else if (stepLength > 60) {
                    sessionAggrAccumulator.add(Constants.STEP_PERIOD_60);
                }
            }

        });
    }


    /**
     * 计算各session范围占比，并写入MySQL
     *
     * @param value
     */
    private static void calculateAndPersistAggrStat(String value, long taskid) {
        // 从Accumulator统计串中获取值
        long session_count = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.SESSION_COUNT));

        long visit_length_1s_3s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1s_3s));
        long visit_length_4s_6s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_4s_6s));
        long visit_length_7s_9s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_7s_9s));
        long visit_length_10s_30s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10s_30s));
        long visit_length_30s_60s = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30s_60s));
        long visit_length_1m_3m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_1m_3m));
        long visit_length_3m_10m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_3m_10m));
        long visit_length_10m_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_10m_30m));
        long visit_length_30m = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.TIME_PERIOD_30m));

        long step_length_1_3 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_1_3));
        long step_length_4_6 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_4_6));
        long step_length_7_9 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_7_9));
        long step_length_10_30 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_10_30));
        long step_length_30_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_30_60));
        long step_length_60 = Long.valueOf(StringUtils.getFieldFromConcatString(
                value, "\\|", Constants.STEP_PERIOD_60));

        // 计算各个访问时长和访问步长的范围
        double visit_length_1s_3s_ratio = NumberUtils.formatDouble(
                (double) visit_length_1s_3s / (double) session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble(
                (double) visit_length_4s_6s / (double) session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble(
                (double) visit_length_7s_9s / (double) session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble(
                (double) visit_length_10s_30s / (double) session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble(
                (double) visit_length_30s_60s / (double) session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble(
                (double) visit_length_1m_3m / (double) session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble(
                (double) visit_length_3m_10m / (double) session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_10m_30m / (double) session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble(
                (double) visit_length_30m / (double) session_count, 2);

        double step_length_1_3_ratio = NumberUtils.formatDouble(
                (double) step_length_1_3 / (double) session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble(
                (double) step_length_4_6 / (double) session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble(
                (double) step_length_7_9 / (double) session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble(
                (double) step_length_10_30 / (double) session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble(
                (double) step_length_30_60 / (double) session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble(
                (double) step_length_60 / (double) session_count, 2);

        // 将统计结果封装为Domain对象
        SessionAggrStat sessionAggrStat = new SessionAggrStat();
        sessionAggrStat.setTaskid(taskid);
        sessionAggrStat.setSession_count(session_count);
        sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
        sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
        sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
        sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
        sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
        sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
        sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
        sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
        sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
        sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
        sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
        sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
        sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
        sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
        sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio);

        // 调用对应的DAO插入统计结果
        ISessionAggrStatDAO sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO();
        sessionAggrStatDAO.insert(sessionAggrStat);
    }

}